package com.chuangke.elasticsearch.service;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import com.alibaba.fastjson.JSON;
import com.chuangke.common.exception.ChuangkeException;
import com.chuangke.elasticsearch.model.ElasticEntity;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.ElasticsearchException;
import co.elastic.clients.elasticsearch.cat.IndicesResponse;
import co.elastic.clients.elasticsearch.cat.indices.IndicesRecord;
import co.elastic.clients.elasticsearch.core.BulkResponse;
import co.elastic.clients.elasticsearch.core.DeleteResponse;
import co.elastic.clients.elasticsearch.core.GetResponse;
import co.elastic.clients.elasticsearch.core.IndexRequest;
import co.elastic.clients.elasticsearch.core.IndexResponse;
import co.elastic.clients.elasticsearch.core.bulk.BulkOperation;
import co.elastic.clients.elasticsearch.indices.DeleteIndexResponse;
import co.elastic.clients.elasticsearch.indices.GetIndexResponse;
import co.elastic.clients.elasticsearch.indices.IndexState;
import co.elastic.clients.transport.endpoints.BooleanResponse;

//@Component
public class ElasticsearchService {

	private static final Logger logger = LoggerFactory.getLogger(ElasticsearchService.class);

	@Autowired
	private ElasticsearchClient elasticsearchClient;

	// ------------------------------------index操作------------------------------------------

	/**
	 * @param index 索引名称
	 * @return void
	 * @throws IOException 
	 * @throws ElasticsearchException 
	 */
	public void createIndex(String index) {
		try {
			elasticsearchClient.indices().create(c -> c.index(index));
		} catch (ElasticsearchException | IOException e) {
			throw new ChuangkeException(e) ;
		}
	}

	/**
	 * 断某个index是否存在
	 * 
	 * @return boolean
	 * @throws IOException 
	 * @throws ElasticsearchException 
	 */
	public boolean isExistsIndex(String indexName) {
		BooleanResponse exists;
		try {
			exists = elasticsearchClient.indices().exists(d -> d.index(indexName));
		} catch (ElasticsearchException | IOException e) {
			throw new ChuangkeException(e) ;
		}
		return exists.value();
	}

	/**
	 * 删除索引
	 * 
	 * @param index
	 * @return
	 * @throws IOException
	 */
	public boolean deleteIndex(String index) {
		DeleteIndexResponse deleteIndexResponse;
		try {
			deleteIndexResponse = elasticsearchClient.indices().delete(s -> s.index(index));
		} catch (ElasticsearchException | IOException e) {
			throw new ChuangkeException(e) ;
		}
		return deleteIndexResponse.acknowledged();
	}

	/**
	 * 获取索引
	 * 
	 * @param index
	 * @return
	 * @throws IOException
	 */
	public Map<String, IndexState> getIndex(String index) throws IOException {
		// 查看指定索引
		GetIndexResponse getIndexResponse = elasticsearchClient.indices().get(s -> s.index(index));
        //        result.forEach((k, v) -> log.info("key = {},value = {}",k ,v));
		return getIndexResponse.result();
	}

	/**
	 * 获取全部索引
	 * 
	 * @return
	 * @throws IOException 
	 * @throws ElasticsearchException 
	 */
	public List<IndicesRecord> getAllIndex() throws ElasticsearchException, IOException {
		// 查看全部索引
		IndicesResponse indicesResponse = elasticsearchClient.cat().indices();
		// 返回对象具体查看 co.elastic.clients.elasticsearch.cat.indices.IndicesRecord
		return indicesResponse.valueBody();
	}

	// ------------------------------------document操作------------------------------------------

	/**
	 * @param entity  对象
	 * @return void
	 * @throws IOException 
	 * @throws ElasticsearchException 
	 */
	public <T> void insertDocument(String index, ElasticEntity<T> entity) {
		logger.info("Data : id={},entity={}", entity.getId(), JSON.toJSONString(entity.getData()));
		IndexRequest<T> request = IndexRequest.of(i -> i.index(index).id(entity.getId()).document(entity.getData()));
		IndexResponse response;
		try {
			response = elasticsearchClient.index(request);
		} catch (ElasticsearchException | IOException e) {
			throw new ChuangkeException(e) ;
		}
		logger.info("Indexed with version " + response.version());
	}

	/**
	 * 更新文档
	 * 
	 * @param <T>
	 * @param index
	 * @param entity
	 * @param clazz
	 * @throws IOException
	 */
	public <T> void updateDocument(String index, ElasticEntity<T> entity, Class<T> clazz) {
		BooleanResponse exists  ;
		try {
			exists = elasticsearchClient.exists(e -> e.index(index).id(entity.getId()));
			if(exists.value()){
				elasticsearchClient
						.update(e -> e.index(index).id(entity.getId()).doc(entity.getData()), clazz);
			}else{
				elasticsearchClient.create(e -> e.index(index).id(entity.getId()).document(entity.getData()));
			}
		} catch (ElasticsearchException | IOException e) {
			throw new ChuangkeException(e) ;
		}
	}

	/**
	 * 获取文档
	 * 
	 * @param <T>
	 * @param index
	 * @param id
	 * @param clazz
	 * @return
	 * @throws IOException
	 */
	public <T> T getDocument(String index, String id, Class<T> clazz) {
		GetResponse<T> getResponse;
		try {
			getResponse = elasticsearchClient.get(s -> s.index(index).id(id), clazz);
		} catch (ElasticsearchException | IOException e) {
			throw new ChuangkeException(e) ;
		}

		if (getResponse.found()) {
			return getResponse.source();
		}
		
		return null ;
	}

	/**
	 * 判断文档是否存在
	 * 
	 * @param index
	 * @param id
	 * @return
	 * @throws IOException
	 */
	public Boolean existDocument(String index, String id) throws IOException {
		BooleanResponse booleanResponse = elasticsearchClient.exists(s -> s.index(index).id(id));
		return booleanResponse.value();
	}

	/**
	 * 删除文档
	 * 
	 * @param index
	 * @param id
	 * @throws IOException
	 */
	public void deleteDocument(String index, String id) {
		DeleteResponse deleteResponse;
		try {
			deleteResponse = elasticsearchClient.delete(s -> s.index(index).id(id));
		} catch (ElasticsearchException | IOException e) {
			throw new ChuangkeException(e) ;
		}
		logger.info("删除文档操作结果:{}", deleteResponse.result());
	}

	/**
	 * 批量插入
	 * 
	 * @param <T>
	 * @param index
	 * @param list
	 * @throws IOException
	 */
	public <T> void insertBatch(String index, List<ElasticEntity<T>> list) {
		List<BulkOperation> bulkOperations = new ArrayList<>();
		list.forEach(
				u -> bulkOperations.add(BulkOperation.of(b -> b.index(c -> c.id(u.getId()).document(u.getData())))));
		BulkResponse bulkResponse;
		try {
			bulkResponse = elasticsearchClient.bulk(s -> s.index(index).operations(bulkOperations));
		} catch (ElasticsearchException | IOException e) {
			throw new ChuangkeException(e) ;
		}
		bulkResponse.items().forEach(i -> logger.info("i = {}", i.result()));
		logger.error("bulkResponse.errors() = {}", bulkResponse.errors());
	}

	/**
	 * 批量删除
	 * @param <T>
	 * @param index
	 * @param list
	 * @throws IOException
	 */
	public <T> void deleteBatch(String index, List<String> list) {
		List<BulkOperation> bulkOperations = new ArrayList<>();
		list.forEach(a -> bulkOperations.add(BulkOperation.of(b -> b.delete(c -> c.id(a)))));
		BulkResponse bulkResponse;
		try {
			bulkResponse = elasticsearchClient.bulk(a -> a.index("users").operations(bulkOperations));
			bulkResponse.items().forEach(a -> logger.info("result = {}", a.result()));
			logger.error("bulkResponse.errors() = {}", bulkResponse.errors());
		} catch (ElasticsearchException | IOException e) {
			throw new ChuangkeException(e) ;
		}
	}


//	/**
//	 * @param idxName
//	 * @param builder
//	 * @return void
//	 */
//	public void deleteByQuery(String idxName, BulkRequest.Builder builder) {
//		DeleteByQueryRequest request = new DeleteByQueryRequest(builder);
//		request.setQuery(builder);
//		// 设置批量操作数量,最大为10000
//		request.setBatchSize(10000);
//		request.setConflicts("proceed");
//		try {
//			elasticsearchClient.deleteByQuery(request);
//		} catch (Exception e) {
//			throw new RuntimeException(e);
//		}
//	}

//	/**
//	 * @param idxName index
//	 * @param builder 查询参数
//	 * @param c       结果类对象
//	 * @return java.util.List<T>
//	 */
//	public <T> List<T> search(String idxName, SearchSourceBuilder builder, Class<T> c) {
//		SearchRequest request = new SearchRequest(idxName);
//		request.source(builder);
//		try {
//			SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT);
//			SearchHit[] hits = response.getHits().getHits();
//			List<T> res = new ArrayList<>(hits.length);
//			for (SearchHit hit : hits) {
//				res.add(JSON.parseObject(hit.getSourceAsString(), c));
//			}
//			return res;
//		} catch (Exception e) {
//			throw new RuntimeException(e);
//		}
//	}

}
